package com.sinozo.data.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.bean.UserPageActionBean;
import com.sinozo.data.bean.UserPageViewBean;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.*;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author zhaoyb
 * @version 1.0
 * @description 用户动作详情
 * @date 2023/12/6 16:03:36
 */
public class DwsPageActionDetailWindow {


    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(DwsUserUserStartWindow.class);
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);

            //开启checkpoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, org.apache.flink.api.common.time.Time.of(3L, TimeUnit.DAYS), org.apache.flink.api.common.time.Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath, "dwsPageActionDetailWindow")));
            env.setStateBackend(new EmbeddedRocksDBStateBackend());

            String topic = ConfigConstant.KAFKA_ACTION_LOG_TOPIC;
            String groupId = ConfigConstant.KAFKA_PAGE_ACTION_DETAIL_GROUP_ID;
            DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

            //3.将每行数据转换为JavaBean对象
            SingleOutputStreamOperator<UserPageActionBean> userPageActionDS = kafkaDS.map(pageView -> {
                JSONObject jsonObject = JSON.parseObject(pageView);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject action = jsonObject.getJSONObject("action");
                Long ts = action.getLong("ts");

                return UserPageActionBean.builder().appVer(common.getString("app_ver"))
                        .os(common.getString("os")).osVersion(common.getString("os_version")).device(common.getString("device"))
                        .deviceBrand(common.getString("device_brand")).isNew(common.getLong("is_new"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).code(common.getString("code")).labCode(common.getString("lab_code"))
                        .labGroupCode(common.getString("lab_group_code")).deviceId(common.getString("device_id"))
                        .createTime(DateFormatUtil.toYmdHms(ts)).ts(System.currentTimeMillis()).event(action.getString("event"))
                        .subEvent(action.getString("sub_event")).actionType(action.getString("action_type")).build();
            });

            //6.将数据写出到ClickHouse
            userPageActionDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_user_page_action_detail values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

            //7.启动任务
            env.execute("DwsPageActionDetailWindow");
        } catch (Exception e) {
            logger.error("Exception", e);
        }
    }


}
